package com.its.common.net.server;

import cn.hutool.core.date.StopWatch;
import cn.hutool.core.lang.Assert;
import com.its.common.net.core.codec.DecoderBuildParams;
import com.its.common.net.server.channel.ChannelHelper;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.socket.DatagramPacket;
import io.netty.util.ReferenceCountUtil;
import io.netty.util.concurrent.DefaultEventExecutorGroup;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Author xiaxp
 * @Date 9/1/22 16:52
 * @Description
 */
@Slf4j
public abstract class BaseServer<T> implements Server<T>{

    protected EventLoopGroup boss;
    protected EventLoopGroup worker;
    protected DefaultEventExecutorGroup defaultEventExecutorGroup;
    protected DecoderBuildParams decoderParams;
    protected List<SimpleChannelInboundHandler> handlers;
    protected Channel channel;
    protected int port;
    protected AtomicBoolean serverStarted = new AtomicBoolean(false);
    /**
     * 读写缓冲区大小2M
     */
    protected final int READ_WRITE_BUF_CAP = 2 * 1024 * 1024;

    private BaseServer(){}

    public BaseServer(int port, List<SimpleChannelInboundHandler> handlers) {
        this.port = port;
        this.handlers = handlers;
    }

    @Override
    public void start() {
        Assert.checkBetween(port, 1, 65535, "端口号区间应该在1~65526");

        if (serverStarted.compareAndSet(false, true)) {
            this.defaultEventExecutorGroup = new DefaultEventExecutorGroup(
                    8,
                    new ThreadFactory() {
                        private AtomicInteger threadIndex = new AtomicInteger(0);
                        @Override
                        public Thread newThread(Runnable r) {
                            return new Thread(r, "NettyServerWorkerThread_" + this.threadIndex.incrementAndGet());
                        }
                    });


            StopWatch stopWatch = new StopWatch();
            stopWatch.start();

            ChannelFuture future = null;
            try {
                future = doStart(port);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            final ChannelFuture finalFuture = future;
            finalFuture.addListener(f -> {
                if (f.isSuccess()) {
                    channel = finalFuture.channel();
                    serverStarted.set(true);

                    stopWatch.stop();
                    long cost = stopWatch.getTotalTimeMillis();
                    log.info("[{}] 服务端开启, 端口号:{}, 花费:{}[ms]", this.getClass().getSimpleName(), port, cost);

                    doAfterStartUp(channel);
                }else{
                    log.error("[{}] 服务端启动失败, 端口号:{}, 原因:{}", this.getClass().getSimpleName(), port,
                            f.cause().getMessage());
                }
            });

            //添加ShutdownHook
            Runtime.getRuntime().addShutdownHook(new Thread(() -> stop()));
        }
    }

    @Override
    public void stop() {
        if (!serverStarted.get()) {
            log.warn("ShutdownHook: Server hasn't started yet!");
            return;
        }
        if(null != boss) {
            boss.shutdownGracefully();
        }
        if(null != worker) {
            worker.shutdownGracefully();
        }
        if(null != defaultEventExecutorGroup) {
            defaultEventExecutorGroup.shutdownGracefully();
        }
    }

    /** 服务端启动操作
     * @param port 端口
     * @return future
     */
    public abstract ChannelFuture doStart(int port) throws InterruptedException;

    /** 服务器启动后的操作
     * @param channel 通道
     */
    public abstract void doAfterStartUp(Channel channel);

    @Override
    public void kickClient(String ipAddr, int port) {
        Channel channel = ChannelHelper.getChannelFromAddrAndPort(ipAddr, port);
        if(null != channel) {
            kickClient(channel.id());
        }
    }

    public void kickClient(ChannelId channelId) {
        ChannelHelper.closeChannel(channelId);
    }

    @Override
    public abstract void sendToAll(ByteBuf content);

    @Override
    public void send(T ch, ByteBuf content) {
        if(ch instanceof Channel){
            Channel channel = (Channel)ch;
            channel.writeAndFlush(content);
        }else if (ch instanceof InetSocketAddress){
            InetSocketAddress addr = (InetSocketAddress) ch;
            DatagramPacket sendPacket = new DatagramPacket(content, addr);
            this.channel.writeAndFlush(sendPacket);
        }

        return;
    }

    @Override
    public void sendToAll(byte[] content) {
        ByteBuf buf = Unpooled.copiedBuffer(content);
        sendToAll(buf);
    }

    @Override
    public void send(T ch, byte[] content) {
        ByteBuf buf = Unpooled.copiedBuffer(content);
        send(ch, buf);
    }
}
